package com.alibaba.rocketmq.example.quickstart;

import java.io.DataOutputStream;
import java.io.File;
import java.io.FileOutputStream;
import java.io.IOException;
import java.util.Collections;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.alibaba.fastjson.JSON;
import com.alibaba.rocketmq.client.exception.MQBrokerException;
import com.alibaba.rocketmq.client.exception.MQClientException;
import com.alibaba.rocketmq.common.UtilAll;
import com.alibaba.rocketmq.common.admin.ConsumeStats;
import com.alibaba.rocketmq.common.message.MessageExt;
import com.alibaba.rocketmq.common.protocol.body.ConsumerConnection;
import com.alibaba.rocketmq.common.protocol.body.QueueTimeSpan;
import com.alibaba.rocketmq.remoting.common.RemotingHelper;
import com.alibaba.rocketmq.remoting.exception.RemotingException;
import com.alibaba.rocketmq.tools.admin.DefaultMQAdminExt;
import com.alibaba.rocketmq.tools.admin.api.MessageTrack;

public class ConsoleMaster {
	public static Logger logger = LoggerFactory.getLogger(ConsoleMaster.class);
	//private String nameServer = "10.230.20.224:9876";
	private String nameServer = "192.168.1.102:9876";
	DefaultMQAdminExt defaultMQAdminExt;
	Throwable t = null;
	Map<String, String> map = new LinkedHashMap<String, String>();

	public ConsoleMaster() {
		 defaultMQAdminExt = getDefaultMQAdminExt();
		 try {
			defaultMQAdminExt.start();
		} catch (MQClientException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}
	public static void main(String[] args) throws Throwable {
		ConsoleMaster consoleMaster = new ConsoleMaster();
		/*
		 * Set<QueueTimeSpan> queryConsumeTimeSpan = consoleMaster.queryConsumeTimeSpan("OMS_WaybillTrack_REQUEST_OUT", "WQS_WaybillTrackOrder_CGROUP"); System.out.println("queryConsumeTimeSpan====" + queryConsumeTimeSpan);
		 */
		/*ConsumeStats consumeStats = consoleMaster.examineConsumeStats("testGroup");
		String jsonString = JSON.toJSONString(consumeStats);
		System.out.println("consumeStats=========" + jsonString);*/
		
		//consoleMaster.resetOffsetByTimestamp("TopicTest", "ConsumerTEST8_GROUP");
		ConsumerConnection examineConsumerConnectionInfo = consoleMaster.examineConsumerConnectionInfo("ConsumerTEST39_GROUP");
		/*
		 * Map<String, String> queryMsgById = consoleMaster.queryMsgById("0AE0423700002A9F0000000000000870"); System.out.println(queryMsgById);
		 */
	}
	public ConsumerConnection examineConsumerConnectionInfo(String consumerGroup) {
		try {
			ConsumerConnection examineConsumerConnectionInfo = defaultMQAdminExt.examineConsumerConnectionInfo(consumerGroup);
			return examineConsumerConnectionInfo;
		} catch (InterruptedException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		} catch (MQBrokerException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		} catch (RemotingException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		} catch (MQClientException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		return null;
	}
	public void resetOffsetByTimestamp(String topic,String consumerGroup) {
		try {
			defaultMQAdminExt.resetOffsetByTimestamp(topic, consumerGroup, System.currentTimeMillis(), true);
		} catch (RemotingException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		} catch (MQBrokerException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		} catch (InterruptedException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		} catch (MQClientException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}
	public ConsumeStats examineConsumeStats(String consumerGroup) throws MQClientException {
		
		try {
			ConsumeStats consumeStats = defaultMQAdminExt.examineConsumeStats(consumerGroup);
			return consumeStats;
		} catch (RemotingException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		} catch (MQClientException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		} catch (InterruptedException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		} catch (MQBrokerException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		return null;
	}
	/*
	 * public Pair<MessageView, List<MessageTrack>> viewMessage(String subject, final String msgId) { try { MQAdminExt mqAdminExt; MessageExt messageExt = mqAdminExt.viewMessage(subject, msgId); List<MessageTrack> messageTrackList = messageTrackDetail(messageExt); return new Pair<>(MessageView.fromMessageExt(messageExt), messageTrackList); } catch (Exception e) { throw Throwables.propagate(e); } }
	 */
	public Set<QueueTimeSpan> queryConsumeTimeSpan(final String topic, final String group) {
		try {
			defaultMQAdminExt = getDefaultMQAdminExt();
			defaultMQAdminExt.start();
			Set<QueueTimeSpan> queryConsumeTimeSpan = defaultMQAdminExt.queryConsumeTimeSpan(topic, group);
			return queryConsumeTimeSpan;
		} catch (InterruptedException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		} catch (MQBrokerException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		} catch (RemotingException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		} catch (MQClientException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		return null;

	}

	public Map<String, String> queryMsgById(String msgId) throws Throwable {

		try {
			defaultMQAdminExt.start();
			MessageExt msg = defaultMQAdminExt.viewMessage(msgId);
			// ConsumeMessageDirectlyResult consumeMessageDirectly = defaultMQAdminExt.consumeMessageDirectly("ConsumerGroup", "", msgId);
			// System.out.println("========="+consumeMessageDirectly);
			List<MessageTrack> messageTrackList = messageTrackDetail(msg);
			System.out.println("====================" + messageTrackList);
			// String bodyTmpFilePath = createBodyFile(msg);
			map.put("Topic", msg.getTopic());
			map.put("Tags", "[" + msg.getTags() + "]");
			map.put("Keys", "[" + msg.getKeys() + "]");
			map.put("Queue ID", String.valueOf(msg.getQueueId()));
			map.put("Queue Offset:", String.valueOf(msg.getQueueOffset()));
			map.put("CommitLog Offset:", String.valueOf(msg.getCommitLogOffset()));
			map.put("Born Timestamp:", UtilAll.timeMillisToHumanString2(msg.getBornTimestamp()));
			map.put("Store Timestamp:", UtilAll.timeMillisToHumanString2(msg.getStoreTimestamp()));
			map.put("Born Host:", RemotingHelper.parseSocketAddressAddr(msg.getBornHost()));
			map.put("Store Host:", RemotingHelper.parseSocketAddressAddr(msg.getStoreHost()));
			map.put("System Flag:", String.valueOf(msg.getSysFlag()));
			map.put("Properties:", msg.getProperties() != null ? msg.getProperties().toString() : "");
			// map.put("Message Body Path:", bodyTmpFilePath);
			System.out.println("==========" + msg.getConsumeTimestamp());
			return map;
		} catch (Throwable e) {
			logger.error(e.getMessage(), e);
			t = e;
		} finally {
			shutdownDefaultMQAdminExt(defaultMQAdminExt);

		}
		throw t;
	}

	@SuppressWarnings("unused")
	private String createBodyFile(MessageExt msg) throws IOException {
		DataOutputStream dos = null;

		try {
			String bodyTmpFilePath = "/tmp/rocketmq/msgbodys";
			File file = new File(bodyTmpFilePath);
			if (!file.exists()) {
				file.mkdirs();
			}
			bodyTmpFilePath = bodyTmpFilePath + "/" + msg.getMsgId();
			dos = new DataOutputStream(new FileOutputStream(bodyTmpFilePath));
			dos.write(msg.getBody());
			return bodyTmpFilePath;
		} finally {
			if (dos != null)
				dos.close();
		}
	}

	protected DefaultMQAdminExt getDefaultMQAdminExt() {
		DefaultMQAdminExt defaultMQAdminExt = new DefaultMQAdminExt();
		defaultMQAdminExt.setNamesrvAddr(nameServer);
		defaultMQAdminExt.setInstanceName(Long.toString(System.currentTimeMillis()));
		return defaultMQAdminExt;
	}

	protected void shutdownDefaultMQAdminExt(DefaultMQAdminExt defaultMQAdminExt) {
		defaultMQAdminExt.shutdown();
	}

	public List<MessageTrack> messageTrackDetail(MessageExt msg) {
		try {
			return defaultMQAdminExt.messageTrackDetail(msg);
		} catch (Exception e) {
			logger.error("op=messageTrackDetailError", e);
			return Collections.emptyList();
		}
	}

	public String getNameServer() {
		return nameServer;
	}

	public void setNameServer(String nameServer) {
		this.nameServer = nameServer;
	}

}
